package com.atguigu.flink0922.chapter07.state;

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/6 14:07
 */
public class Flink02_State_Operator_Broadcast {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment
            .getExecutionEnvironment()
            .setParallelism(3);
        // 数据流
        DataStreamSource<String> dataStream = env.socketTextStream("hadoop162", 9999);
        // 控制流
        final BroadcastStream<String> controlStream = env
            .socketTextStream("hadoop162", 8888)
            .broadcast(new MapStateDescriptor<String, String>("bd_state", String.class, String.class));
        
        dataStream
            .connect(controlStream)
            .process(new BroadcastProcessFunction<String, String, String>() {
                @Override
                public void processElement(String value,
                                           ReadOnlyContext ctx,
                                           Collector<String> out) throws Exception {
                    final ReadOnlyBroadcastState<String, String> state = ctx
                        .getBroadcastState(new MapStateDescriptor<>("bd_state", String.class, String.class));
                    
                     String aSwitch = state.get("switch");
                    aSwitch = aSwitch == null ? "0" : aSwitch;
                    switch (aSwitch) {
                        case "1":
                            out.collect("处理模式切换到 1号逻辑");
                            break;
                        case "2":
                            out.collect("处理模式切换到 2号逻辑");
                            break;
                        default:
                            out.collect("处理模式切换到默认逻辑");
                            break;
                    }
                    
                }
                
                @Override
                public void processBroadcastElement(String value,  // 0 1 2
                                                    Context ctx,
                                                    Collector<String> out) throws Exception {
                    ctx
                        .getBroadcastState(new MapStateDescriptor<>("bd_state", String.class, String.class))
                        .put("switch", value);
                    
                }
            })
            .print();
        
        env.execute();
        
    }
}
